/**
 * Date:2022年1月6日 上午10:58:20
 * Copyright (c) 2022, www.bwbroad.com All Rights Reserved
 */

package com.app.common.core;

import java.util.HashMap;
import java.util.Objects;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.support.Acknowledgment;

import com.app.common.vo.DebeziumMessage;
import com.app.common.vo.DebeziumMessage.OpType;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectReader;

/**
 * Description: 抽象消费者 <br/>
 * Date:2022年12月6日 上午10:58:20 <br/>
 * 
 * @author xuejianxin
 */
public abstract class AbstractDebeziumConsumer {
	private static org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(AbstractDebeziumConsumer.class);

	private static final ObjectMapper jsonMapper = new ObjectMapper();
	private static final ObjectReader jsonReader = jsonMapper.reader();

	protected <T, ID> void consumer(ConsumerRecord<String, String> message, Acknowledgment ack, IMessageHandler h) {
		log.info(">>>>>>>topic={},key={}, consume message: {}", message.topic(), message.key(), message.value());
		// 1.解析
		DebeziumMessage msgObj = parseMessage(message.value(), message.key(), message.topic());
		// 2.处理消息
		boolean success = h.handleMessage(msgObj);
		if (success) {
			// 3. 应答
			ack.acknowledge();
		}
	}

	@SuppressWarnings("unchecked")
	private final DebeziumMessage parseMessage(String message, String key, String topic) {
		DebeziumMessage msgObj = null;
		try {

			// 获取topic 第一个点之前的子串，如果没有点 则取全部
			int prifixIndex = topic.indexOf(".");
			prifixIndex = prifixIndex > 0 ? prifixIndex : topic.length();
			String topicPrefix = topic.substring(0, prifixIndex);
			msgObj = new DebeziumMessage();
			msgObj.setTopicPrefix(topicPrefix);
			// 1.解析key
			boolean needParseKey = false;
			if (needParseKey) {// 目前看起来key 没啥用，先不解析
				JsonNode keyRoot = jsonReader.readTree(key);
				JsonNode keyPayload = keyRoot.path("payload");
				HashMap<String, Object> keyMap = jsonReader.treeToValue(keyPayload, HashMap.class);
				msgObj.setKey(keyMap);
			}

			// 2.解析value
			if (!Objects.isNull(message)) {// 删除的墓碑消息没有消息体，只有key
				JsonNode root = jsonReader.readTree(message);
				JsonNode payload = root.path("payload");
				String opStr = payload.path("op").asText();// 获取操作
				JsonNode source = payload.path("source");
				String table = source.path("table").asText();
				String file = source.path("file").asText();
				String pos = source.path("pos").asText();

				HashMap<String, Object> beforeMap = jsonReader.treeToValue(payload.get("before"), HashMap.class);
				HashMap<String, Object> afterMap = jsonReader.treeToValue(payload.get("after"), HashMap.class);
				msgObj.setOp(OpType.valueOf(opStr));
				msgObj.setTable(table);
				msgObj.setFile(file);
				msgObj.setPos(pos);
				msgObj.setBefore(beforeMap);
				msgObj.setAfter(afterMap);
			} else {// 删除的墓碑消息没有消息体，只有key
				msgObj.setOp(OpType.e);
			}
			// 3.解析数据类型,不知道数据类型的情况下可能无法格式化日期，不需要的时候可以省掉这个步骤
			// ...后边处理,20221222 不需要处理了

		} catch (Exception e) {
			log.error(topic, e);
		}
		return msgObj;
	}

}
